package com.blyat.xsoft.kernel.component.mq.rocketmq.consumer.push;

import com.blyat.xsoft.kernel.component.mq.MQConsumer;
import com.blyat.xsoft.kernel.component.mq.rocketmq.RocketMQContextLoader;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.DisposableBean;

/**
 * @Auther: syh
 * @Date: 2020/8/7
 * @Description:
 */
abstract class PushConsumer<T> extends RocketMQContextLoader
        implements MQConsumer<T>, DisposableBean {
    private final static int DEFAULT_MAX_RECONSUME_TIMES = 2;
    private final static String DEFAULT_TAGS = "*";
    private final static int DEFAULT_BATCH_SIZE = 32;
    private static final int DEFAULT_IO_THREADS = Math.min(Runtime.getRuntime().availableProcessors() + 1, 32);

    private String groupName;
    private String topic;
    private String tags;
    private int maxReconsumeTimes;
    private int threadMax;
    private int threadMin;
    private int batchSize;

    private DefaultMQPushConsumer consumer;

    public PushConsumer(String topic) {
        this(topic, DEFAULT_BATCH_SIZE);
    }

    public PushConsumer(String topic, int batchSize) {
        this(topic, topic, batchSize);
    }

    public PushConsumer(String groupName, String topic, int batchSize) {
        this(groupName, topic, DEFAULT_TAGS, batchSize);
    }

    public PushConsumer(String groupName, String topic, String tags, int batchSize) {
        this(groupName, topic, tags, batchSize, DEFAULT_MAX_RECONSUME_TIMES);
    }

    public PushConsumer(String groupName, String topic, String tags, int batchSize, int maxReconsumeTimes) {
        this(groupName, topic, tags, batchSize, maxReconsumeTimes, DEFAULT_IO_THREADS, DEFAULT_IO_THREADS);
    }

    public PushConsumer(String groupName, String topic, String tags, int batchSize, int maxReconsumeTimes, int threadMax, int threadMin) {
        this.groupName = groupName;
        this.topic = topic;
        this.tags = tags;
        this.batchSize = batchSize;
        this.maxReconsumeTimes = maxReconsumeTimes;
        this.threadMax = threadMax;
        this.threadMin = threadMin;
    }

    @Override
    protected void doAfterSingletonsInstantiated() {
        createRocketMQConsumer();
    }

    private void createRocketMQConsumer() {
        try {
            consumer = new DefaultMQPushConsumer(groupName);
            consumer.setNamesrvAddr(getProperties().getNameServer());
            consumer.setConsumeThreadMax(threadMax);
            consumer.setConsumeThreadMin(threadMin);
            // todo 超过32无效，如果需要设置大于32的值，
            //  需修改rocketmq broker配置并重启: maxTransferCountOnMessageInMemory
            consumer.setConsumeMessageBatchMaxSize(batchSize);
            consumer.setPullBatchSize(batchSize);
            /*
             * CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，跳过历史消息
             * CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
             */
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            // 默认是集群方式，可以更改为广播，但是广播方式不支持重试
            consumer.setMessageModel(MessageModel.CLUSTERING);
            consumer.setMaxReconsumeTimes(maxReconsumeTimes);
            consumer.subscribe(topic, tags);
            consumer.registerMessageListener(this);
            consumer.start();
        } catch (Exception e) {
            //
        }
    }

    @Override
    public void destroy() throws Exception {
        if (this.consumer != null) {
            this.consumer.shutdown();
            ;
        }
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public void setTags(String tags) {
        this.tags = tags;
    }

    public void setMaxReconsumeTimes(int maxReconsumeTimes) {
        this.maxReconsumeTimes = maxReconsumeTimes;
    }

    public void setThreadMax(int threadMax) {
        this.threadMax = threadMax;
    }

    public void setThreadMin(int threadMin) {
        this.threadMin = threadMin;
    }

    public void setBatchSize(int batchSize) {
        this.batchSize = batchSize;
    }
}
